package streating
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import java.sql.{Connection, DriverManager, PreparedStatement}

object s4 {
  def main(args: Array[String]) {
    // Spark配置
    val conf = new SparkConf()
      .setAppName("SocketToMySQL")
      .setMaster("local[2]")
    // 创建StreamingContext，批次间隔5秒
    val ssc = new StreamingContext(conf, Seconds(5))
    // 从socket源创建DStream
    val lines = ssc.socketTextStream("localhost", 9999)
    // 处理数据：拆分单词并计数
    val wordCounts = lines.flatMap(_.split(" "))
                          .filter(_.nonEmpty)
                          .map(word => (word, 1))
                          .reduceByKey(_ + _)
    // 将结果保存到MySQL
    wordCounts.foreachRDD { rdd =>
      rdd.foreachPartition { partitionOfRecords =>
        var connection: Connection = null
        var preparedStatement: PreparedStatement = null
        try {
          // MySQL连接参数
          val url = "jdbc:mysql://localhost:3306/spark_db"
          val username = "root"
          val password = "123456"
          // 加载JDBC驱动
          Class.forName("com.mysql.jdbc.Driver")
          // 建立连接
          connection = DriverManager.getConnection(url, username, password)
          // 准备SQL语句（使用ON DUPLICATE KEY UPDATE实现更新）
          val sql =
            """
              |INSERT INTO word_counts (word, count)
              |VALUES (?, ?)
              |ON DUPLICATE KEY UPDATE count = count + VALUES(count)
              |"""
          preparedStatement = connection.prepareStatement(sql)
          // 遍历分区中的记录
          partitionOfRecords.foreach { case (word, count) =>
            preparedStatement.setString(1, word)
            preparedStatement.setInt(2, count)
            preparedStatement.executeUpdate()
          }
        }
        catch {
          case e: Exception => e.printStackTrace()
        } finally {
          // 关闭资源
          if (preparedStatement != null) preparedStatement.close()
          if (connection != null) connection.close()
        }
      }
    }
    // 启动流计算
    ssc.start()
    ssc.awaitTermination()
  }
}
